package org.mq.mymq.nameserver.event.listener;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.NodeAckDto;
import org.mq.mymq.common.enums.NameServerEventCode;
import org.mq.mymq.common.enums.NameServerResponseCode;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.enums.ReplicationMsgTypeEnum;
import org.mq.mymq.nameserver.event.model.NodeReplicationAckMsgEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @ClassName NodeReplicationAckMsgEventListener
 * @Description 接收上一个节点同步完成的ack信号
 * @Author jiarong_ye
 * @Date 2025/6/25 12:16
 * @Version 1.0
 */
public class NodeReplicationAckMsgEventListener implements Listener<NodeReplicationAckMsgEvent> {
    private static final Logger log = LoggerFactory.getLogger(NodeReplicationAckMsgEventListener.class);

    @Override
    public void onReceive(NodeReplicationAckMsgEvent event) throws Exception {
        boolean isHeadNode = CommonCache.getPreNodeChannel() == null;
        if (isHeadNode) {
            // 如果是头节点，告诉broker消息同步完成
            NodeAckDto nodeAckDto = CommonCache.getNodeAckMap().get(event.getMsgId());
            Channel brokerChannel = nodeAckDto.getBrokerCtx().channel();
            if (!brokerChannel.isActive()) {
                throw new RuntimeException("brokerChannel is not active");
            }
            CommonCache.getNodeAckMap().remove(event.getMsgId());
            if (ReplicationMsgTypeEnum.REGISTRY.getCode() == event.getType()) {
                brokerChannel.writeAndFlush(new TcpMsg(NameServerResponseCode.REGISTRY_SUCCESS.getCode(), NameServerResponseCode.REGISTRY_SUCCESS.getDesc().getBytes()));
            } else if (ReplicationMsgTypeEnum.HEART_BEAT.getCode() == event.getType()) {
                brokerChannel.writeAndFlush(new TcpMsg(NameServerResponseCode.HEART_BEAT_SUCCESS.getCode(), NameServerResponseCode.HEART_BEAT_SUCCESS.getDesc().getBytes()));
            }
            log.info("发送消息同步完成信号到broker");
        } else {
            // 如果是中间节点，转发ack信号
            CommonCache.getPreNodeChannel().writeAndFlush(new TcpMsg(NameServerEventCode.NODE_REPLICATION_ACK_MSG.getCode(),
                    JSON.toJSONBytes(event)));
            log.info("转发ack信号");
        }
    }
}
